package com.example.core.raft;

import com.example.core.utils.IdUtils;
import com.example.core.utils.RandomUtils;
import org.jgroups.*;
import org.jgroups.blocks.executor.ExecutionService;
import org.jgroups.protocols.raft.RAFT;
import org.jgroups.protocols.raft.Role;
import org.jgroups.raft.RaftHandle;
import org.jgroups.raft.blocks.CounterService;
import org.jgroups.raft.blocks.ReplicatedStateMachine;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

/*
 * @className: Server
 * @author: chentao
 * @date: 2025/2/21 下午2:38
 * @Version: 1.0
 * @description:
 */
public class RaftServer implements RAFT.RoleChange {

    Logger logger = Logger.getLogger(RaftServer.class.getName());

    public static final String DEFAULT_SERVER = "DEFAULT_SITC_SERVER";

    private String serverId;

    private View view;

    private RaftListener raftListener = new RaftListener() {

        @Override
        public void memberJoin(Address address) {
            logger.warning("节点: " + address.toString() + " 已加入到集群...");
        }

        @Override
        public void memberLeave(Address address) {
            logger.warning("节点: " + address.toString() + " 已离开到集群...");
        }

        @Override
        public void receive(Message message) {
            logger.warning("收到消息: " + message.getObject());
        }

        @Override
        public void isLeader(String serverId) {
            if (raftHandle.isLeader()) {
                scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable() {
                    @Override
                    public void run() {
                        if (raftHandle.isLeader()) {
                            logger.warning("主节点: " + serverId + " 执行业务任务");
                        } else {
                            if (scheduledThreadPoolExecutor != null && !scheduledThreadPoolExecutor.isShutdown()) {
                                scheduledThreadPoolExecutor.shutdown();
                            }
                        }
                    }
                }, 0, rate, TimeUnit.SECONDS);
            }
        }
    };

    private ThreadPoolExecutor threadPoolExecutor;

    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

    private Receiver receiver = new Receiver() {

        //收到集群消息
        @Override
        public void receive(Message message) {
            threadPoolExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    if (raftListener != null) {
                        raftListener.receive(message);
                    }
                }
            });
        }

        //判断
        @Override
        public void viewAccepted(View new_view) {
            if (view != null) {
                List<Address> leftMembers = View.leftMembers(view, new_view);
                List<Address> joinedMembers = View.newMembers(view, new_view);
                if (view.getMembers().size() > new_view.getMembers().size()) {
                    //有节点掉线了
                    if (!leftMembers.isEmpty()) {
                        for (Address address : leftMembers) {
                            threadPoolExecutor.execute(new Runnable() {
                                @Override
                                public void run() {
                                    if (raftListener != null) {
                                        raftListener.memberLeave(address);
                                    }
                                }
                            });
                        }
                    }
                } else {
                    if (!joinedMembers.isEmpty()) {
                        for (Address address : joinedMembers) {
                            threadPoolExecutor.execute(new Runnable() {
                                @Override
                                public void run() {
                                    if (raftListener != null) {
                                        raftListener.memberJoin(address);
                                    }
                                }
                            });
                        }
                    }
                }
            }
            view = new_view;
        }
    };

    private JChannel jChannel;

    private RaftHandle raftHandle;

    private CounterService counterService;

    private ExecutionService executionService;

    private int corePoolSize = 2;

    private int maxPoolSize = 10;

    private int keepAliveTime = 60;

    private int rate = 5;

    private TimeUnit timeUnit = TimeUnit.SECONDS;

    private ReplicatedStateMachine replicatedStateMachine;

    public RaftServer() {
        this.serverId = IdUtils.getId();
    }

    public RaftServer(String serverId) {
        this.serverId = serverId;
    }

    public RaftServer(String serverId, RaftListener raftListener) {
        this.serverId = serverId;
        this.raftListener = raftListener;
    }

    /***
     * 启动集群
     * @throws Exception
     */
    public void start() throws Exception {
        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(corePoolSize);
        this.threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, timeUnit, new LinkedBlockingQueue<>());
        this.jChannel = new JChannel("jgroups/udp.xml").name(serverId);
        this.replicatedStateMachine = new ReplicatedStateMachine(this.jChannel);
        this.counterService = new CounterService(this.jChannel);
        this.raftHandle = new RaftHandle(this.jChannel, this.replicatedStateMachine).raftId(serverId);
        this.raftHandle.raft().setMembers(this.serverId);
        this.raftHandle.addRoleListener(this);
        this.executionService = new ExecutionService(this.jChannel);
        this.jChannel.setReceiver(this.receiver);
        this.jChannel.connect(DEFAULT_SERVER);
        logger.warning("server " + serverId + " started ...");
    }

    /***
     * 关闭集群
     * @throws Exception
     */
    public void stop() throws Exception {
        this.executionService.shutdownNow();
        if (this.jChannel != null) {
            this.jChannel.disconnect();
            this.jChannel.close();
        }
        this.threadPoolExecutor.shutdownNow();
        this.scheduledThreadPoolExecutor.shutdownNow();
    }

    /****
     * 集群内推送信息
     * @param message
     * @throws Exception
     */
    public Object sendAll(Object message) throws Exception {
        this.jChannel.send(null, message);
        return message;
    }

    /***
     * 单节点发送消息
     * @param address
     * @param message
     * @return
     * @throws Exception
     */
    public Object send(Address address, Object message) throws Exception {
        this.jChannel.send(address, message);
        return message;
    }

    /***
     * 获取集群活跃的节点列表
     * @return
     */
    public List<Address> getServers() throws Exception {
        return this.jChannel.getView().getMembers();
    }


    /****
     * 随机获取一个节点
     * @return
     * @throws Exception
     */
    public Address getServer() throws Exception {
        return RandomUtils.getAddress(this.jChannel);
    }

    /***
     * 获取当前节点
     * @return
     * @throws Exception
     */
    public Address getSelfServer() throws Exception {
        return this.jChannel.getAddress();
    }

    /****
     * 发生了选举
     * @param role
     */
    @Override
    public void roleChanged(Role role) {
        if (role == Role.Leader) {
            this.threadPoolExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    if (raftListener != null) {
                        raftListener.isLeader(serverId);
                    }
                }
            });
        }
    }

    /***
     * 监听类
     */
    interface RaftListener {

        public void memberJoin(Address address);

        public void memberLeave(Address address);

        public void receive(Message message);

        public void isLeader(String serverId);
    }

    ///////////////////////////


    public String getServerId() {
        return serverId;
    }

    public void setServerId(String serverId) {
        this.serverId = serverId;
    }

    public View getView() {
        return view;
    }

    public void setView(View view) {
        this.view = view;
    }

    public ThreadPoolExecutor getThreadPoolExecutor() {
        return threadPoolExecutor;
    }

    public void setThreadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {
        this.threadPoolExecutor = threadPoolExecutor;
    }

    public Receiver getReceiver() {
        return receiver;
    }

    public void setReceiver(Receiver receiver) {
        this.receiver = receiver;
    }

    public JChannel getjChannel() {
        return jChannel;
    }

    public void setjChannel(JChannel jChannel) {
        this.jChannel = jChannel;
    }

    public void setRaftHandle(RaftHandle raftHandle) {
        this.raftHandle = raftHandle;
    }

    public CounterService getCounterService() {
        return counterService;
    }

    public void setCounterService(CounterService counterService) {
        this.counterService = counterService;
    }

    public ExecutionService getExecutionService() {
        return executionService;
    }

    public void setExecutionService(ExecutionService executionService) {
        this.executionService = executionService;
    }

    public int getCorePoolSize() {
        return corePoolSize;
    }

    public void setCorePoolSize(int corePoolSize) {
        this.corePoolSize = corePoolSize;
    }

    public int getMaxPoolSize() {
        return maxPoolSize;
    }

    public void setMaxPoolSize(int maxPoolSize) {
        this.maxPoolSize = maxPoolSize;
    }

    public int getKeepAliveTime() {
        return keepAliveTime;
    }

    public void setKeepAliveTime(int keepAliveTime) {
        this.keepAliveTime = keepAliveTime;
    }

    public TimeUnit getTimeUnit() {
        return timeUnit;
    }

    public void setTimeUnit(TimeUnit timeUnit) {
        this.timeUnit = timeUnit;
    }

    public void setReplicatedStateMachine(ReplicatedStateMachine replicatedStateMachine) {
        this.replicatedStateMachine = replicatedStateMachine;
    }

    public RaftHandle getRaftHandle() {
        return raftHandle;
    }

    public ReplicatedStateMachine getReplicatedStateMachine() {
        return replicatedStateMachine;
    }

    public RaftListener getRaftListener() {
        return raftListener;
    }

    public void setRaftListener(RaftListener raftListener) {
        this.raftListener = raftListener;
    }

    public ScheduledThreadPoolExecutor getScheduledThreadPoolExecutor() {
        return scheduledThreadPoolExecutor;
    }

    public void setScheduledThreadPoolExecutor(ScheduledThreadPoolExecutor scheduledThreadPoolExecutor) {
        this.scheduledThreadPoolExecutor = scheduledThreadPoolExecutor;
    }

    public int getRate() {
        return rate;
    }

    public void setRate(int rate) {
        this.rate = rate;
    }
}
